package xyz.codej.websocket.netty.server.handler;

import cn.hutool.core.collection.CollectionUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.HttpRequestHandler;
import xyz.codej.websocket.netty.server.config.NettyServerConfig;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

/**
 * @ClassName: WebSocketServerHandler
 * @Description TODO
 * @Author 王华君
 * @Date 2020/4/1 3:29 下午
 * @VERSION 1.0
 */
@Slf4j
@Component
// 使用时每次创建新的handler
//@Scope("prototype")
// 共享handler
@ChannelHandler.Sharable
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<Object> {

    @Autowired
    NettyServerConfig serverConfig;

    /**
     * websocket 服务的 uri
     */
    private String wsUri;

    private ChannelGroup group;

    private ConcurrentHashMap<String, ChannelId> clientChannelMap;

    private WebSocketServerHandshaker handShaker;

    private final static int HTTP_STATUS_OK = 200;

    private String socketUrl;

    @PostConstruct
    void init() {
        this.socketUrl = "ws://" + serverConfig.getAddress() + ":" + serverConfig.getPort() + wsUri;
    }

    public ChannelGroup getGroup() {
        return group;
    }

    public TextWebSocketFrameHandler setGroup(ChannelGroup group) {
        this.group = group;
        return this;
    }

    public ConcurrentHashMap<String, ChannelId> getClientChannelMap() {
        return clientChannelMap;
    }

    public TextWebSocketFrameHandler setClientChannelMap(ConcurrentHashMap<String, ChannelId> clientChannelMap) {
        this.clientChannelMap = clientChannelMap;
        return this;
    }

    public String getWsUri() {
        return wsUri;
    }

    public TextWebSocketFrameHandler setWsUri(String wsUri) {
        this.wsUri = wsUri;
        return this;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //当WebSocket连接成功
        if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
            //不需要再处理HTTP请求
            ctx.pipeline().remove(String.valueOf(HttpRequestHandler.class));
            //告诉已经连接的客户端有新的客户端连接进来了
            group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel() + " joined"));
            //将新的客户端添加到ChannelGroup中
            group.add(ctx.channel());
        } else {
            super.userEventTriggered(ctx, evt);
        }

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 传统的HTTP接入
        if (msg instanceof FullHttpRequest) {
            log.info("传统的HTTP接入");
            handleHandShakerHttpRequest(ctx, (FullHttpRequest) msg);
        }
        // WebSocket接入
        else if (msg instanceof TextWebSocketFrame) {
            log.info("WebSocket接入");
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    /**
     * 每当从服务端收到新的客户端连接时，客户端的 Channel 存入 ChannelGroup 列表中
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端连接");
        group.add(ctx.channel());
        super.handlerAdded(ctx);
    }

    /**
     * 每当从服务端收到客户端断开时，客户端的 Channel 自动从 ChannelGroup 列表中移除了
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端断开连接");
//        group 中自动会把 channel 移除
//        group.forEach(channel -> log.info(channel.id().asLongText()));

        super.handlerRemoved(ctx);
    }

    /**
     * 服务端监听到客户端活动
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        log.info("客户端加入:" + incoming.remoteAddress());
        super.channelActive(ctx);
    }

    /**
     * 服务端监听到客户端不活动
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        log.info("客户端离线:" + incoming.remoteAddress());
        super.channelInactive(ctx);
    }

    /**
     * 处理握手的http请求
     *
     * @param ctx
     * @param req
     * @throws Exception
     */
    private void handleHandShakerHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {

        // 如果HTTP解码失败，返回HTTP异常
        if (!req.decoderResult().isSuccess()
                || (!"websocket".equals(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
            return;
        }

        // 构造握手响应返回，本机测试
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(this.socketUrl, null, false);

        handShaker = wsFactory.newHandshaker(req);
        if (handShaker == null) {
            WebSocketServerHandshakerFactory
                    .sendUnsupportedVersionResponse(ctx.channel());
        } else {
            // 握手成功
            handShaker.handshake(ctx.channel(), req);

            // 保存请求和Channel关系
            QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
            Map<String, List<String>> parameters = queryStringDecoder.parameters();
            List<String> reqId = parameters.get("reqId");

            if (CollectionUtil.isNotEmpty(reqId)) {
                log.info("客户端：{} 握手成功", reqId.get(0));
                clientChannelMap.put(reqId.get(0), ctx.channel().id());
            }
        }
    }

    /**
     * websocket 请求
     *
     * @param ctx
     * @param frame
     */
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // 判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handShaker.close(ctx.channel(),
                    (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否是Ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(
                    new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 本例程仅支持文本消息，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format(
                    "%s frame types not supported", frame.getClass().getName()));
        }

        TextWebSocketFrame textWebSocketFrame = ((TextWebSocketFrame) frame);
        Channel incoming = ctx.channel();
        // 返回应答消息
        String request = textWebSocketFrame.text();
        log.info(String.format("%s received %s", incoming, request));
        incoming.write(
                // 如果是向客户端发送文本消息，则需要发送 TextWebSocketFrame 消息
                new TextWebSocketFrame(request
                        + " , 欢迎使用Netty WebSocket服务，现在时刻："
                        + new java.util.Date().toString()));
    }

    /**
     * 返回应答给客户端
     *
     * @param ctx
     * @param req
     * @param res
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
        if (res.status().code() != HTTP_STATUS_OK) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(),
                    CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
            HttpUtil.setContentLength(res, res.content().readableBytes());
        }

        // 如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpUtil.isKeepAlive(req) || res.status().code() != HTTP_STATUS_OK) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}